package org.demo;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @author yexi
 */
public class KafkaProducerDemo {
    public static void main(String[] args) throws InterruptedException {
        boolean isSetJAAS = false;

        //1.创建链接参数
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        if (isSetJAAS) {
            // 设置 JAAS 配置属性
            props.put("security.protocol", "SASL_PLAINTEXT");
            props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
            String jaasConfig = "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"%s\" password=\"%s\";";
            jaasConfig = String.format(jaasConfig, "kafka", "kafka-secret");
            props.put(SaslConfigs.SASL_JAAS_CONFIG, jaasConfig);
        }

        //2.创建生产者
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            System.out.println("成功创建Kafka生产者");
            //3.发送至消息队列
            for (int i = 0; /*i < 10*/ ; i++) {
                Thread.sleep(1000);
                System.out.println("准备发送");
                ProducerRecord<String, String> record = new ProducerRecord<>("topic03", "key" + i, "value" + i);
                producer.send(record);
                System.out.println("发送");
            }

        }
    }
}
